package com.huan.gateway.flow;

import java.util.concurrent.Flow;

public class MySubscriber implements Flow.Subscriber {
    /**
     * 和发布者订阅关系对象
     */
    private Flow.Subscription subscription;

    /**
     * 订阅消息
     * @param subscription a new subscription
     */
    @Override
    public void onSubscribe(Flow.Subscription subscription) {

        //1. 保持订阅关系
        this.subscription = subscription;

        //2. 告诉发布者，我要订阅一个消息，发布者就会把消息推送到下面的 onNext 方法里面
        this.subscription.request(1);

    }

    /**
     * 消费订阅的信息
     * @param item the item
     */
    @Override
    public void onNext(Object item) {
        // 1. 接受数据
        System.out.println("item = " + item);

//                throw new RuntimeException("测试异常");
        // 2.处理完成数据之后，我们可以告诉发布者我还要订阅一个消息，程序就以此运行下去
        this.subscription.request(1);
        // 2.1 或者也可以告诉发布者。我不在接受数据了,那么 onComplete 方法就不会调用。
//                this.subscription.cancel();

    }

    /**
     *  消费消息出现异常会进到这步
     * @param throwable the exception
     */
    @Override
    public void onError(Throwable throwable) {
        System.out.println(throwable.getMessage());
        //出现异常，我们就可以告诉发布者。不接受数据了
        this.subscription.cancel();
    }

    /**
     * 在全部数据消费完成，这个方法才会调用。如果中途消费时候调用 cancel 方法，这个方法就不会触发了
     */
    @Override
    public void onComplete() {
        System.out.println("MySubscriber 订阅者全部数据消费完成 ");
    }
}
